#include <memory>

#include "leveldb/db.h"
#include "leveldb/options.h"
#include "netlib/base/logger.h"
#include "netlib/net/event/event_loop.h"
#include "netlib/net/socket/inet_address.h"
#include "netlib/rpc/rpc_server.h"

#include "kvstore.pb.h"

namespace netlib::examples {

class KVStoreServiceImpl : public KVStoreService {
public:
	explicit KVStoreServiceImpl(std::string_view db_path = "./db") {
		leveldb::Options opts;
		opts.create_if_missing = true;
		opts.write_buffer_size = 4 * 1024;
		leveldb::DB* db = nullptr;
		auto s = db_->Open(opts, static_cast<std::string>(db_path), &db);
		if (!s.ok()) {
			LOG_FATAL << "Failed to open leveldb in " << db_path;
		}
		db_.reset(db);
	}

	void Put(::google::protobuf::RpcController* controller,
	         const ::netlib::examples::PutRequest* request,
	         ::netlib::examples::Response* response,
	         ::google::protobuf::Closure* done) override {
		auto s = db_->Put(write_opts_, request->key(), request->value());
		response->set_success(s.ok());
		if (!s.ok()) {
			response->set_message(s.ToString());
		}
		LOG_DEBUG << "Put " << request->key() << (s.ok() ? " successfully" : "unsuccessfully");
		if (done) {
			done->Run();
		}
	}
	void Get(::google::protobuf::RpcController* controller,
	         const ::netlib::examples::GetRequest* request,
	         ::netlib::examples::Response* response,
	         ::google::protobuf::Closure* done) override {
		std::string value;
		auto s = db_->Get(read_opts_, request->key(), &value);
		response->set_success(s.ok());
		if (s.ok()) {
			response->set_value(value);
		} else {
			response->set_message(s.ToString());
		}
		if (done) {
			done->Run();
		}
	}
	void Remove(::google::protobuf::RpcController* controller,
	            const ::netlib::examples::RemoveRequest* request,
	            ::netlib::examples::Response* response,
	            ::google::protobuf::Closure* done) override {
		auto s = db_->Delete(write_opts_, request->key());
		response->set_success(s.ok());
		if (!s.ok()) {
			response->set_message(s.ToString());
		}
		if (done) {
			done->Run();
		}
	}

private:
	std::unique_ptr<leveldb::DB> db_;
	leveldb::WriteOptions write_opts_;
	leveldb::ReadOptions read_opts_;
};

class KVServer : NonCopyable {
public:
	KVServer(net::EventLoop* loop,
	         const net::InetAddress& addr,
	         std::string_view db_path = "./db") :
	    service_(db_path),
	    server_(loop, addr, false, "KVServer") {
		server_.Register(&service_);
	}

	void Start() { server_.Start(); }

private:
	KVStoreServiceImpl service_;
	rpc::RpcServer server_;
};

} // namespace netlib::examples

int main() {
	using namespace netlib;
	net::InetAddress addr(13589);
	net::EventLoop loop;
	examples::KVServer server(&loop, addr);
	server.Start();
	loop.Loop();
}
